package com.partsoft.dits.dsm;

import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.partsoft.dits.EdgeException;
import com.partsoft.dits.MonitorIntegrateDescriptor;
import com.partsoft.dits.MonitorStatus;
import com.partsoft.dits.TableDataEdgeDescriptor;
import com.partsoft.dits.log.Log;
import com.partsoft.dits.message.DataAction;
import com.partsoft.dits.message.DataProperty;
import com.partsoft.dits.utils.DbUtils;
import com.partsoft.dits.utils.ExceptionUtils;
import com.partsoft.dits.utils.ExecutionCallback;
import com.partsoft.dits.utils.JdbcTemplate;
import com.partsoft.dits.utils.ReturnExecutionCallback;

import org.springframework.jdbc.CannotGetJdbcConnectionException;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.util.StringUtils;

public class DataSourceMonitorTask implements Runnable {

	private DataSourceMonitorContext monitorContext;

	private MonitorStatus monitorStatus = MonitorStatus.STOPPED;

	private long monitorInterval = MonitorIntegrateDescriptor.DEFAULT_MONITOR_INTERVAL;

	private boolean removeAfterRetrieve = true;

	private int retrieveCount = MonitorIntegrateDescriptor.DEFAULT_RETRIEVE_COUNT;

	private DataSourceMonitorDescriptor monitorDescriptor;

	DataSourceMonitorTask() {
	}

	DataSourceMonitorTask(DataSourceMonitorDescriptor desc, DataSourceMonitorContext context) {
		setMonitorDescriptor(desc);
		setMonitorInterval(desc.getMonitorInterval());
		setRemoveAfterRetrieve(desc.isRemoveAfterRetrieved());
		setRetrieveCount(desc.getRetrieveCount());
		setMonitorContext(context);
	}

	public DataSourceMonitorDescriptor getMonitorDescriptor() {
		return monitorDescriptor;
	}

	public void setMonitorDescriptor(DataSourceMonitorDescriptor monitorDescriptor) {
		this.monitorDescriptor = monitorDescriptor;
	}

	public int getRetrieveCount() {
		return retrieveCount;
	}

	public void setRetrieveCount(int retrieveCount) {
		this.retrieveCount = retrieveCount;
	}

	public long getMonitorInterval() {
		return monitorInterval;
	}

	public void setMonitorInterval(long monitorInterval) {
		this.monitorInterval = monitorInterval;
	}

	public void run() {
		setMonitorStatus(MonitorStatus.STARTED);
		boolean errorFocus = false;
		while (getMonitorStatus() != MonitorStatus.STOPPED) {
			if (getMonitorStatus() == MonitorStatus.PAUSED) {
				try {
					Thread.sleep(200);
				} catch (Throwable e) {
				}
				continue;
			}
			long interValCount = (1000 * getMonitorInterval()) / 200;
			boolean interValCon = false;
			while (interValCount-- > 0) {
				try {
					Thread.sleep(200);
				} catch (Exception e) {
				}
				if (getMonitorStatus() == MonitorStatus.STOPPED || getMonitorStatus() == MonitorStatus.PAUSED) {
					interValCon = true;
					break;
				}
			}
			
			if (interValCon) continue;
			synchronized (this) {
				try {
					doMonitor();
					if (errorFocus) {
						Log.warn(new EdgeException(monitorDescriptor, "数据库连接恢复正常"));
					}
					errorFocus = false;
				} catch (Throwable e) {
					if (ExceptionUtils.findCause(e, CannotGetJdbcConnectionException.class) != null) {
						if (!errorFocus) {
							errorFocus = true;
							Log.error(new EdgeException(monitorDescriptor, new StringBuilder("数据库连接发生错误，错误信息：").append(
									e.getMessage()).toString(), e));
						}
					} else {
						Log.error(new EdgeException(monitorDescriptor, e.getMessage(), e));
					}
				}
			}
		}
		setMonitorStatus(MonitorStatus.STOPPED);
	}

	public DataSourceMonitorContext getMonitorContext() {
		return monitorContext;
	}

	void setMonitorContext(DataSourceMonitorContext context) {
		this.monitorContext = context;
	}

	protected void doMonitor() throws Exception {
		final DataSourceMonitorContext ctx = getMonitorContext();
		final JdbcTemplate jdbchelper = new JdbcTemplate(ctx.getDataSource());

		List<RecordChangeHistory> messagelist = DbUtils.executeInReadonly(ctx.getDataSource(),
				new ReturnExecutionCallback<List<RecordChangeHistory>>() {
					public List<RecordChangeHistory> callback() {
						jdbchelper.setMaxRows(getRetrieveCount());
						return new ArrayList<RecordChangeHistory>(
								jdbchelper.query(
										String.format(
												"SELECT SID, NSM, CLTB, OPT, CLID, INUSE, OPTM FROM %s WHERE INUSE=1 AND NSM='%s' ORDER BY SID",
												ctx.getMonitorLogTable(), ctx.getMonitorLogNS()),
										new DataMonitorMessageRowMapper()));
					}
				});

		if (messagelist.size() == 0)
			return;

		final StringBuffer updateSql = new StringBuffer();
		if (isRemoveAfterRetrieve()) {
			updateSql.append("DELETE FROM " + ctx.getMonitorLogTable() + " WHERE SID in (");
		} else {
			updateSql.append("UPDATE " + ctx.getMonitorLogTable() + " SET INUSE = 0 WHERE SID in (");
		}
		boolean tag = false;
		for (RecordChangeHistory dmmsg : messagelist) {
			if (tag) {
				updateSql.append("," + dmmsg.getSid());
			} else {
				updateSql.append(dmmsg.getSid());
			}
			tag = true;

			String[] fsNames = StringUtils.split(dmmsg.getTableName(), ".");
			if (fsNames == null || fsNames.length != 2) {
				Log.warn(new EdgeException(getMonitorDescriptor(), String.format("获取的变更记录(SID:%d)失败，表名(%s)不正确。",
						dmmsg.getSid(), dmmsg.getTableName())));
				continue;
			}

			TableDataEdgeDescriptor descriptors[] = ctx.getEdgeDescriptor(fsNames[0], fsNames[1]);
			if (descriptors.length == 0) {
				Log.warn(new EdgeException(getMonitorDescriptor(), String.format("获取变更记录(SID:%d)，表名(%s)找不到发布配置。",
						dmmsg.getSid(), dmmsg.getTableName())));
				continue;
			}

			Map<DataProperty, String> keyMaps = new HashMap<DataProperty, String>(1);
			keyMaps.put(descriptors[0].getKeyColumn(), dmmsg.getKeyData());

			List<Map<DataProperty, String>> changedRows = DbUtils.getTableRecordsByKey(ctx.getDataSource(), fsNames[0],
					fsNames[1], keyMaps, true, dmmsg.getOption());
			for (TableDataEdgeDescriptor descriptor : descriptors) {
				try {
					ctx.processMonitorData(descriptor, dmmsg.getOption(), changedRows);
				} catch (Throwable e) {
					Log.error(String.format("处理表数据发布(%s)时出错: %s", descriptors.toString(), e.getMessage()), e);
				}
			}
		}

		updateSql.append(")");

		DbUtils.executeInTransaction(ctx.getDataSource(), TransactionDefinition.PROPAGATION_REQUIRES_NEW,
				new ExecutionCallback() {
					public void callback() {
						jdbchelper.update(updateSql.toString());
					}
				});
	}

	public void stop() {
		this.setMonitorStatus(MonitorStatus.STOPPED);
	}

	public void pause() {
		setMonitorStatus(MonitorStatus.PAUSED);
	}

	public void resume() {
		setMonitorStatus(MonitorStatus.STARTED);
	}

	public void setMonitorStatus(MonitorStatus statu) {
		synchronized (this) {
			monitorStatus = statu;
		}
	}

	public MonitorStatus getMonitorStatus() {
		synchronized (this) {
			return monitorStatus;
		}
	}

	public boolean isRemoveAfterRetrieve() {
		return removeAfterRetrieve;
	}

	public void setRemoveAfterRetrieve(boolean removeAfterRetrieve) {
		this.removeAfterRetrieve = removeAfterRetrieve;
	}

	private static class DataMonitorMessageRowMapper implements RowMapper<RecordChangeHistory> {
		public RecordChangeHistory mapRow(ResultSet rs, int rowNum) throws SQLException {
			RecordChangeHistory monmessage = new RecordChangeHistory();
			monmessage.setSid(rs.getLong("SID"));
			monmessage.setTableName(rs.getString("CLTB"));
			monmessage.setTableOption(DataAction.fromCode(rs.getInt("OPT")));
			monmessage.setKeyData(rs.getString("CLID"));
			monmessage.setEnabled(rs.getBoolean("INUSE"));
			monmessage.setOptionTime(rs.getTimestamp("OPTM"));
			return monmessage;
		}
	}
}
